package stream;

import org.apache.flink.api.common.functions.FlatMapFunction;
import org.apache.flink.api.java.ExecutionEnvironment;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.connectors.redis.RedisSink;
import org.apache.flink.streaming.connectors.redis.common.config.FlinkJedisPoolConfig;
import org.apache.flink.streaming.connectors.redis.common.mapper.RedisCommand;
import org.apache.flink.streaming.connectors.redis.common.mapper.RedisCommandDescription;
import org.apache.flink.streaming.connectors.redis.common.mapper.RedisMapper;
import org.apache.flink.util.Collector;


public class SinkRedis {
    public static void main(String[] args) throws Exception {
    //    创建运行环境
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
    //    创建输入数据源
        DataStreamSource<String> source = env.socketTextStream("192.168.43.51", 1234);

        DataStream<addcountwithworld> result = source.flatMap(new FlatMapFunction<String, addcountwithworld>() {
            public void flatMap(String s, Collector<addcountwithworld> collector) throws Exception {
                String[] list_ = s.split(" ");
                System.out.println(s);
                //System.out.println(list_);
                System.out.println("---------");
                for (String world : list_) {
                    collector.collect(new addcountwithworld(world, 1));
                }
            }
        }).keyBy("world").sum("count");
        //result.print();

        //创建redis sink
        FlinkJedisPoolConfig conf = new FlinkJedisPoolConfig.Builder().setHost("192.168.217.134").setPort(6379).build();

        RedisSink<addcountwithworld> sink = new RedisSink<addcountwithworld>(conf, new RedisMapper<addcountwithworld>() {
            public RedisCommandDescription getCommandDescription() {
                return new RedisCommandDescription(RedisCommand.SET );
            }

            public String getKeyFromData(addcountwithworld addcountwithworld) {
                return addcountwithworld.world;
            }

            public String getValueFromData(addcountwithworld addcountwithworld) {
                return String.valueOf(addcountwithworld.count);
            }
        });

        // result保存到sink中
        result.addSink(sink);


        env.execute("rediessink");


    }

    public static class addcountwithworld{
    //    定义两个变量
        private String world;
        private int count;
        public addcountwithworld(){}
        public addcountwithworld(String world, int count){
            this.world = world;
            this.count = count;
        }

        public String getWorld() {
            return world;
        }

        public void setWorld(String world) {
            this.world = world;
        }

        public int getCount() {
            return count;
        }

        public void setCount(int count) {
            this.count = count;
        }



        @Override
        public String toString() {
            return "addcountwithworld{" +
                    "world='" + world + '\'' +
                    ", count=" + count +
                    '}';
        }
    }
}
